package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo03DSL {

  def main(args: Array[String]): Unit = {
    /**
     * SQL中常见的操作：
     * 1、基础操作：
     * select  分组字段
     * ,聚合操作
     * ,使用函数对字段进行处理
     * from tb
     * where 过滤
     * group by 分组
     * having 对分组聚合的结果做过滤
     * order by 排序，asc升序、desc降序
     * limit 限制返回的条数
     *
     * 2、多表操作：
     * 2.1 union联接：表与表之间进行纵向（行的方向）的合并
     * union ：去重
     * union all：不会去重
     *
     * 2.2 join关联：表与表之间进行横向（列的方向）的合并
     * inner内连接：join / inner join
     * outer外连接：
     * 左外连接：left join / left outer join
     * 右外连接：right join / right outer join
     * 全外连接：full join / full outer join
     *
     * 3、高级操作：
     * 行列转换：
     * 一行转多行：lateral view + UDTF函数
     * 多行转一行：group by + collect_list / collect_set
     * 自定义函数：UDF、UDTF、UDAF（基本不用）
     *
     * 4、函数的使用：
     * 字符串函数：substring、split、concat、concat_ws......
     * 数值函数：round、ceil、floor、abs......
     * 条件函数：if、case when、coalesce
     * 日期时间函数：from_unixtime、unix_timestamp、datediff、date_add......
     * 聚合函数：max、min、sum、count、avg
     * 窗口函数：row_number、rank、dense_rank、lag、lead、max、min、sum、count、avg......
     *
     */

    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName.replace("$", ""))
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    // 需要先导入函数包
    import org.apache.spark.sql.functions._

    val stuDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/stu/students.txt")

    // DSL的用法，参考SQL的基本用法学习

    // select
    stuDF
      // 字符串表达式
      .select("id", "name", "clazz")
    //      .show()

    stuDF
      // 列表达式
      .select($"id" as "newId", $"name", $"clazz")
    //      .show()

    // 使用函数，例如：将班级的 文科/理科 提取出来
    stuDF
      .select($"id", $"name", substring($"clazz", 1, 2) as "clz")
    //      .show()

    stuDF
      .select($"id", $"name", row_number() over Window.partitionBy($"clazz").orderBy($"id") as "rn")
    //      .show()

    // where 用于过滤
    // 过滤出理科三班的男同学
    stuDF
      .where($"clazz" === "理科三班" and $"gender" === "男")
    //      .show()

    // filter也可使用类似RDD的方式进行过滤
    stuDF
      .filter(row => {
        var flag: Boolean = false
        val clazz: String = row.getAs[String]("clazz")
        val gender: String = row.getAs[String]("gender")
        if (clazz == "理科二班" && gender == "男") {
          flag = true
        }
        flag
      })
    //      .show()

    // 分组 聚合
    // 统计班级人数
    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id"))
    //      .show()

    // 统计班级人数并按人数降序排列 并返回人数最多的前三个班级
    stuDF
      .groupBy($"clazz")
      .agg(countDistinct($"id") as "cnt")
      .where($"cnt" > 90)
      .orderBy($"cnt".desc)
      .limit(3)
    //      .show()

    // 关联
    val scoDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      .schema("id String,subject_id String,score Int")
      .load("spark/data/stu/score.txt")

    // inner join
    // 关联的字段名相同时不能直接用列表达式，会出现ambiguous问题
    /**
     * 解决方式：
     * 1、对某个DF中的关联字段进行重命名
     * 2、换一种方式指定关联条件
     */
    scoDF
      // 1、对某个DF中的关联字段进行重命名
      .withColumnRenamed("id", "sid")
      .join(stuDF, $"sid" === $"id", "inner")
    //      .show()

    // 2、换一种方式指定关联条件
    scoDF
      .join(stuDF, "id") // 通过字符串指定，但无法再指定关联的方式，默认使用inner join的方式
    //      .show()

    scoDF
      .join(stuDF, List("id"), "inner") // 通过List传入关联的字段，同时也可以指定关联的方式
    //      .show()

    // outer join
    scoDF.join(stuDF, List("id"), "left")
    //      .show()
    scoDF.join(stuDF, List("id"), "right")
    //      .show()
    scoDF.join(stuDF, List("id"), "full")
      .orderBy($"id")
    //      .show()

    // union

    val sampleDF01: DataFrame = stuDF.sample(withReplacement = false, 0.01, 1)
    val sampleDF02: DataFrame = stuDF.sample(withReplacement = false, 0.01, 1)

    sampleDF01
      // 相当于union all没有去重
      .union(sampleDF02)
    //      .show()

    sampleDF01
      .union(sampleDF02)
      .distinct()
    //      .show()

    // 行列转换
    val lineDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "#")
      .schema("line String")
      .load("spark/data/words/words.txt")
    // 一行转多行：将words.txt数据中的每一行切开来，将每个单词转换成一行
    lineDF
      .select(explode(split($"line", ",")) as "word")
    //      .show()

    // 加载weight.txt数据，共三列：id、name、weights
    val weightDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("id String,name String,weights String")
      .load("spark/data/weight.txt")

    // 将weightDF中的每一条数据变成多条，体重有多少个记录就变成多少条，最终输出id name weight
    // 注意：在Hive中需要使用lateral view实现，Spark中可以直接Select

    weightDF.createOrReplaceTempView("weightTB")

    spark.sql(
      """
        |select id
        |       ,name
        |       ,weight
        |from weightTB
        |lateral view explode(split(weights,",")) v as weight
        |""".stripMargin)
    //      .show() // 类似Hive的方式

    spark.sql(
      """
        |select id
        |       ,name
        |       ,explode(split(weights,",")) as weight
        |from weightTB
        |""".stripMargin)
    //      .show()

    val newWeightDF: DataFrame = weightDF
      .select($"id", $"name", explode(split($"weights", ",")) as "weight")

    //    newWeightDF.show()

    // 将处理好的weight数据变回原样
    // 多行转一行：groupBy结合collect_list/collect_set
    val sourceWeightDF: DataFrame = newWeightDF
      .groupBy($"id", $"name")
      .agg(concat_ws(",", collect_list($"weight")) as "weights")

    sourceWeightDF.printSchema()

    sourceWeightDF
      .show()

    // 班级总分Top3
    scoDF
      .groupBy($"id")
      .agg(sum($"score") as "sum_score")
      .join(stuDF, List("id"), "inner")
      .select($"id", $"name", $"clazz", $"sum_score")
      .select($"id", $"name", $"clazz", $"sum_score"
        , row_number() over Window.partitionBy($"clazz").orderBy($"sum_score".desc) as "rn")
      .where($"rn" <= 3)
      .show()

    scoDF
      .groupBy($"id")
      .agg(sum($"score") as "sum_score")
      .join(stuDF, List("id"), "inner")
      .select($"id", $"name", $"clazz", $"sum_score")
      // 给数据增加一列，原有的字段都会保留下来，所以非常适合结合窗口使用
      .withColumn("rn", row_number() over Window.partitionBy($"clazz").orderBy($"sum_score".desc))
      .where($"rn" <= 3)
      .show()

  }

}
